package com.alex.quickstart;

import com.alex.constants.Const;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

/**
 * 生产者
 */
@Slf4j
public class Producer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.指定生产者组的名称
        DefaultMQProducer producer = new DefaultMQProducer("test_quick_start_producer");

        //2.指定nameserver地址
        producer.setNamesrvAddr(Const.NAMESRV_ADDR);

        //3.启动
        producer.start();

        //4.发送消息
        for (int i = 0; i < 5; i++) {
            //4.1创建消息
            Message message = new Message("test-quick-topic", //生产者消息发向哪个topic
                    "TagA", // 标签，主要用于消费端过滤消息
                    "keyA" + i, // 用户自定义的key，唯一的标识
                    ("Hello RocketMQ" + i).getBytes()); // 发送真正的数据

            //4.2生产者发送消息(同步)
            //SendResult sendResult = producer.send(message);
            //log.info("Send Result: {}", sendResult);

            //4.3异步发送消息
            /*producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("Send Result: {}", sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    log.error("Send Error", throwable);
                }
            });*/

            // 4.4 向指定的Queue发送消息 arg:对应于匿名函数的arg参数
            producer.send(message, (messageQueues, message1, arg) -> {
                Integer queueNum = (Integer) arg;
                return messageQueues.get(queueNum);
            },  2);
        }

        //5.关闭生产者(异步的时候, 需要去除)
        producer.shutdown();
    }
}
